SOLID Principles
# Most engineers write this and think it's fine.
class UserManager:
def register(self, email: str, password: str) -> dict:
# validate email
import re
if not re.match(r"[^@]+@[^@]+\.[^@]+", email):
raise ValueError("Invalid email")
# hash password
import hashlib
hashed = hashlib.sha256(password.encode()).hexdigest()
# save to DB
import sqlite3
conn = sqlite3.connect("users.db")
conn.execute("INSERT INTO users VALUES (?, ?)", (email, hashed))
conn.commit()
# send welcome email
import smtplib
server = smtplib.SMTP("smtp.example.com")
server.quit()
# log the event
import logging
logging.info(f"User registered: {email}")
return {"email": email, "status": "registered"}
Spot everything that's wrong. The method reaches into five different domains - validation, crypto, persistence, email, and logging - all in one place. Every time any of those concerns changes, you touch this one class. Every test requires a real SMTP server and a real database. This is SOLID-violating code that looks perfectly reasonable at first glance.
SOLID gives you five precision tools for cutting this kind of problem apart.
What You Will Learn
- Why the Single Responsibility Principle is about reasons to change, not method count
- How Open/Closed applies to Python using strategy registries and
functools.singledispatch - Why Python's structural subtyping with
Protocolsidesteps entire categories of LSP violations - How to break fat ABCs into composable
Protocolinterfaces - How to wire dependency injection manually in Python without a framework, and when a framework pays off
- How to apply all five principles in a single end-to-end FastAPI service
Prerequisites
- Comfortable with Python classes, inheritance, and ABCs (
abc.ABC,@abstractmethod) - Familiar with
dataclasses,typing.Protocol, and type annotations - Basic understanding of FastAPI or any web framework
- Have read Lessons 01–03 (creational, structural, and behavioral patterns)
Part 1 - Single Responsibility Principle (SRP)
A class should have only one reason to change.
Robert Martin's formulation is subtle. He doesn't say "one method" or "one feature." He says one reason to change - one stakeholder, one concern. The UserManager above has at least five: the security team (hashing), the email team (SMTP), the DBA (schema), the product team (validation rules), and the platform team (logging format).
The Violation
# srp/bad.py
import re
import hashlib
import sqlite3
import smtplib
import logging
from typing import Optional
class UserManager:
"""Does everything. Belongs to everyone. Owned by no one."""
def __init__(self, db_path: str, smtp_host: str):
self.db_path = db_path
self.smtp_host = smtp_host
self.logger = logging.getLogger(__name__)
def register(self, email: str, password: str) -> dict:
# Responsibility 1: Input validation (owned by product rules)
if not re.match(r"[^@]+@[^@]+\.[^@]+", email):
raise ValueError(f"Invalid email: {email}")
if len(password) < 8:
raise ValueError("Password too short")
# Responsibility 2: Security (owned by security team)
import secrets
salt = secrets.token_hex(16)
hashed = hashlib.sha256((password + salt).encode()).hexdigest()
# Responsibility 3: Persistence (owned by DBA)
conn = sqlite3.connect(self.db_path)
conn.execute(
"INSERT INTO users (email, password_hash, salt) VALUES (?, ?, ?)",
(email, hashed, salt),
)
conn.commit()
conn.close()
# Responsibility 4: Notification (owned by marketing/email team)
server = smtplib.SMTP(self.smtp_host)
server.sendmail(
email,
f"Subject: Welcome\n\nHi {email}, welcome!",
)
server.quit()
# Responsibility 5: Observability (owned by platform team)
self.logger.info("user.registered", extra={"email": email})
return {"email": email, "status": "registered"}
def authenticate(self, email: str, password: str) -> Optional[str]:
# Responsibility 6: Auth logic mixed in with registration logic
conn = sqlite3.connect(self.db_path)
row = conn.execute(
"SELECT password_hash, salt FROM users WHERE email = ?", (email,)
).fetchone()
conn.close()
if not row:
return None
stored_hash, salt = row
attempt = hashlib.sha256((password + salt).encode()).hexdigest()
if attempt == stored_hash:
import jwt, time
return jwt.encode({"sub": email, "exp": time.time() + 3600}, "secret")
return None
What changes force a rewrite?
| Change | Which stakeholder | Impact |
|---|---|---|
| Switch hashing to bcrypt | Security team | Touch UserManager |
| Move to SendGrid API | Email team | Touch UserManager |
| Migrate DB to PostgreSQL | DBA | Touch UserManager |
| Add phone validation | Product team | Touch UserManager |
| Change log format to JSON | Platform team | Touch UserManager |
| Switch to OAuth JWT | Auth team | Touch UserManager |
Every team in the organisation owns one file. That is the definition of an SRP violation.
The Refactored Design
# srp/good.py
from __future__ import annotations
import re
import hashlib
import secrets
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
# ── Value object ──────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class EmailAddress:
value: str
def __post_init__(self) -> None:
if not re.match(r"[^@]+@[^@]+\.[^@]+", self.value):
raise ValueError(f"Invalid email: {self.value!r}")
def __str__(self) -> str:
return self.value
# ── Responsibility 1: Validation ──────────────────────────────────────────────
class UserValidator:
"""Owned by: product team. Reason to change: validation rules."""
MIN_PASSWORD_LENGTH = 8
def validate_password(self, password: str) -> None:
if len(password) < self.MIN_PASSWORD_LENGTH:
raise ValueError(
f"Password must be at least {self.MIN_PASSWORD_LENGTH} characters"
)
# ── Responsibility 2: Security ────────────────────────────────────────────────
@dataclass(frozen=True)
class HashedPassword:
hash: str
salt: str
class PasswordHasher:
"""Owned by: security team. Reason to change: hashing algorithm."""
def hash(self, plaintext: str) -> HashedPassword:
salt = secrets.token_hex(16)
digest = hashlib.sha256((plaintext + salt).encode()).hexdigest()
return HashedPassword(hash=digest, salt=salt)
def verify(self, plaintext: str, stored: HashedPassword) -> bool:
attempt = hashlib.sha256((plaintext + stored.salt).encode()).hexdigest()
return attempt == stored.hash
# ── Responsibility 3: Persistence ────────────────────────────────────────────
@dataclass
class UserRecord:
email: str
password_hash: str
salt: str
class UserRepository(ABC):
"""Owned by: DBA / data team. Reason to change: storage technology."""
@abstractmethod
def save(self, record: UserRecord) -> None: ...
@abstractmethod
def find_by_email(self, email: str) -> Optional[UserRecord]: ...
class SQLiteUserRepository(UserRepository):
def __init__(self, db_path: str) -> None:
import sqlite3
self._db = db_path
with sqlite3.connect(db_path) as conn:
conn.execute(
"CREATE TABLE IF NOT EXISTS users "
"(email TEXT PRIMARY KEY, password_hash TEXT, salt TEXT)"
)
def save(self, record: UserRecord) -> None:
import sqlite3
with sqlite3.connect(self._db) as conn:
conn.execute(
"INSERT INTO users VALUES (?, ?, ?)",
(record.email, record.password_hash, record.salt),
)
def find_by_email(self, email: str) -> Optional[UserRecord]:
import sqlite3
with sqlite3.connect(self._db) as conn:
row = conn.execute(
"SELECT email, password_hash, salt FROM users WHERE email = ?",
(email,),
).fetchone()
if row:
return UserRecord(*row)
return None
# ── Responsibility 4: Notification ───────────────────────────────────────────
class EmailService(ABC):
"""Owned by: email/marketing team. Reason to change: email provider."""
@abstractmethod
def send_welcome(self, to: EmailAddress) -> None: ...
class SMTPEmailService(EmailService):
def __init__(self, host: str, sender: str) -> None:
self._host = host
self._sender = sender
def send_welcome(self, to: EmailAddress) -> None:
import smtplib
with smtplib.SMTP(self._host) as server:
server.sendmail(
self._sender, str(to), f"Subject: Welcome\n\nHi {to}, welcome!"
)
# ── Responsibility 5: Orchestration (the only thing AuthService does) ─────────
class AuthService:
"""
Owned by: auth team.
Reason to change: registration or authentication workflow.
Delegates every other concern to injected collaborators.
"""
def __init__(
self,
validator: UserValidator,
hasher: PasswordHasher,
repo: UserRepository,
email_svc: EmailService,
logger: logging.Logger,
) -> None:
self._validator = validator
self._hasher = hasher
self._repo = repo
self._email_svc = email_svc
self._log = logger
def register(self, email: str, password: str) -> dict:
addr = EmailAddress(email) # validates format
self._validator.validate_password(password)
hashed = self._hasher.hash(password)
self._repo.save(
UserRecord(
email=str(addr),
password_hash=hashed.hash,
salt=hashed.salt,
)
)
self._email_svc.send_welcome(addr)
self._log.info("user.registered", extra={"email": str(addr)})
return {"email": str(addr), "status": "registered"}
Now each class has one stakeholder and one reason to change. Swapping SMTP for SendGrid means editing only SMTPEmailService. Migrating to PostgreSQL means writing a new PostgresUserRepository - zero other files change.
Part 2 - Open/Closed Principle (OCP)
Software entities should be open for extension, closed for modification.
You extend by adding new code; you don't extend by editing existing code.
The Violation - if/elif Shipping Calculator
# ocp/bad.py
from dataclasses import dataclass
from decimal import Decimal
@dataclass
class Order:
weight_kg: float
destination_country: str
carrier: str # "fedex" | "ups" | "dhl" | "usps"
def calculate_shipping(order: Order) -> Decimal:
"""Every new carrier = edit this function = risk breaking existing carriers."""
if order.carrier == "fedex":
base = Decimal("5.00")
per_kg = Decimal("1.20")
return base + per_kg * Decimal(str(order.weight_kg))
elif order.carrier == "ups":
base = Decimal("4.50")
per_kg = Decimal("1.10")
international_surcharge = (
Decimal("15.00") if order.destination_country != "US" else Decimal("0")
)
return base + per_kg * Decimal(str(order.weight_kg)) + international_surcharge
elif order.carrier == "dhl":
per_kg = Decimal("2.00")
return per_kg * Decimal(str(order.weight_kg))
elif order.carrier == "usps":
if order.weight_kg <= 0.5:
return Decimal("3.50")
return Decimal("3.50") + Decimal("0.80") * Decimal(str(order.weight_kg))
else:
raise ValueError(f"Unknown carrier: {order.carrier!r}")
Adding Royal Mail means opening this function and risking a regression in FedEx logic. The function is open for modification but should be closed for it.
Refactored - Strategy Registry
# ocp/good.py
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from typing import Callable, Protocol
@dataclass(frozen=True)
class Order:
weight_kg: float
destination_country: str
# ── Port (the abstraction that never changes) ─────────────────────────────────
class ShippingStrategy(Protocol):
def calculate(self, order: Order) -> Decimal: ...
# ── Registry (extension point - add without modifying) ───────────────────────
_REGISTRY: dict[str, ShippingStrategy] = {}
def register_carrier(name: str) -> Callable[[type], type]:
"""Class decorator that registers a carrier strategy."""
def decorator(cls: type) -> type:
_REGISTRY[name] = cls()
return cls
return decorator
def calculate_shipping(order: Order, carrier: str) -> Decimal:
"""Never changes, regardless of how many carriers are added."""
strategy = _REGISTRY.get(carrier)
if strategy is None:
raise ValueError(f"Unknown carrier: {carrier!r}")
return strategy.calculate(order)
# ── Concrete strategies (open for addition, never modified after release) ─────
@register_carrier("fedex")
class FedExStrategy:
BASE = Decimal("5.00")
PER_KG = Decimal("1.20")
def calculate(self, order: Order) -> Decimal:
return self.BASE + self.PER_KG * Decimal(str(order.weight_kg))
@register_carrier("ups")
class UPSStrategy:
BASE = Decimal("4.50")
PER_KG = Decimal("1.10")
INTERNATIONAL = Decimal("15.00")
def calculate(self, order: Order) -> Decimal:
surcharge = self.INTERNATIONAL if order.destination_country != "US" else Decimal("0")
return self.BASE + self.PER_KG * Decimal(str(order.weight_kg)) + surcharge
@register_carrier("dhl")
class DHLStrategy:
PER_KG = Decimal("2.00")
def calculate(self, order: Order) -> Decimal:
return self.PER_KG * Decimal(str(order.weight_kg))
@register_carrier("usps")
class USPSStrategy:
BASE = Decimal("3.50")
PER_KG = Decimal("0.80")
def calculate(self, order: Order) -> Decimal:
if order.weight_kg <= 0.5:
return self.BASE
return self.BASE + self.PER_KG * Decimal(str(order.weight_kg))
# Adding Royal Mail: create a new file, register it. Zero existing files change.
# @register_carrier("royal_mail")
# class RoyalMailStrategy: ...
Python-Specific OCP: functools.singledispatch
Python's singledispatch is OCP built into the standard library. You define a base implementation and extend it for new types without touching the original function.
# ocp/singledispatch_example.py
from __future__ import annotations
from functools import singledispatch
from dataclasses import dataclass
from decimal import Decimal
@dataclass(frozen=True)
class StandardPackage:
weight_kg: float
@dataclass(frozen=True)
class FragilePackage:
weight_kg: float
insurance_value: Decimal
@dataclass(frozen=True)
class HazmatPackage:
weight_kg: float
hazmat_class: int
@singledispatch
def shipping_cost(package) -> Decimal:
raise NotImplementedError(f"No shipping rule for {type(package).__name__}")
@shipping_cost.register(StandardPackage)
def _(package: StandardPackage) -> Decimal:
return Decimal("5.00") + Decimal("1.20") * Decimal(str(package.weight_kg))
@shipping_cost.register(FragilePackage)
def _(package: FragilePackage) -> Decimal:
base = Decimal("5.00") + Decimal("1.20") * Decimal(str(package.weight_kg))
insurance = package.insurance_value * Decimal("0.02")
return base + insurance
# Adding hazmat in a NEW file - zero changes to existing code:
@shipping_cost.register(HazmatPackage)
def _(package: HazmatPackage) -> Decimal:
return Decimal("25.00") + Decimal("3.50") * Decimal(str(package.weight_kg))
# Usage
pkg = FragilePackage(weight_kg=2.5, insurance_value=Decimal("500"))
print(shipping_cost(pkg)) # 11.00
Part 3 - Liskov Substitution Principle (LSP)
Subtypes must be substitutable for their base types.
If code that works with Shape must be rewritten to handle Rectangle differently from Circle, LSP is violated. Formally: a subtype must honour the contracts (preconditions, postconditions, invariants) of its parent.
The Classic Violation - Rectangle and Square
# lsp/bad.py
class Rectangle:
def __init__(self, width: float, height: float) -> None:
self._width = width
self._height = height
@property
def width(self) -> float:
return self._width
@width.setter
def width(self, value: float) -> None:
self._width = value
@property
def height(self) -> float:
return self._height
@height.setter
def height(self, value: float) -> None:
self._height = value
def area(self) -> float:
return self._width * self._height
class Square(Rectangle):
"""A square IS a rectangle - in geometry. Not in software."""
@Rectangle.width.setter # type: ignore[override]
def width(self, value: float) -> None:
self._width = value
self._height = value # keep it square!
@Rectangle.height.setter # type: ignore[override]
def height(self, value: float) -> None:
self._width = value # keep it square!
self._height = value
def assert_area(shape: Rectangle, w: float, h: float) -> None:
"""This contract works for Rectangle. Square breaks it silently."""
shape.width = w
shape.height = h
expected = w * h
actual = shape.area()
assert actual == expected, f"Expected {expected}, got {actual}"
r = Rectangle(3, 4)
assert_area(r, 5, 10) # ✓ 50.0
s = Square(3, 3)
assert_area(s, 5, 10) # ✗ AssertionError: Expected 50, got 100
# ^ Square secretly made width=10 too
Square strengthens a postcondition that Rectangle clients don't expect. It is not substitutable.
Python's Solution - Protocol and Structural Subtyping
The real fix is to model behaviour correctly rather than inheritance:
# lsp/good.py
from __future__ import annotations
from typing import Protocol
from dataclasses import dataclass
class Shape(Protocol):
"""Structural protocol - no inheritance required."""
def area(self) -> float: ...
def perimeter(self) -> float: ...
@dataclass(frozen=True)
class Rectangle:
width: float
height: float
def area(self) -> float:
return self.width * self.height
def perimeter(self) -> float:
return 2 * (self.width + self.height)
@dataclass(frozen=True)
class Square:
side: float
def area(self) -> float:
return self.side ** 2
def perimeter(self) -> float:
return 4 * self.side
@dataclass(frozen=True)
class Circle:
radius: float
def area(self) -> float:
import math
return math.pi * self.radius ** 2
def perimeter(self) -> float:
import math
return 2 * math.pi * self.radius
def total_area(shapes: list[Shape]) -> float:
"""Works for any Shape - no isinstance, no special cases."""
return sum(s.area() for s in shapes)
shapes: list[Shape] = [Rectangle(3, 4), Square(5), Circle(2)]
print(total_area(shapes)) # 12.0 + 25.0 + 12.566...
No inheritance means no LSP violation. Square can never accidentally break Rectangle's contract because they share no hierarchy.
A Real LSP Violation - ReadOnlyList
# lsp/readonly_list_bad.py
class ReadOnlyList(list):
"""Inherits list but refuses mutations. Classic LSP violation."""
def append(self, item):
raise TypeError("ReadOnlyList is immutable")
def extend(self, items):
raise TypeError("ReadOnlyList is immutable")
def __setitem__(self, index, value):
raise TypeError("ReadOnlyList is immutable")
def process(items: list) -> None:
items.append(99) # Any list should support this
print(items)
data = ReadOnlyList([1, 2, 3])
process(data) # RuntimeError - broke the list contract
ReadOnlyList inherits list (a supertype) but narrows its capabilities. Code expecting list can no longer use it safely. The correct approach:
# lsp/readonly_list_good.py
from __future__ import annotations
from typing import Protocol, Sequence, TypeVar, overload
T = TypeVar("T", covariant=True) # type: ignore[misc]
class ReadableSequence(Protocol[T]):
"""Protocol that only promises reading - no append contract at all."""
def __len__(self) -> int: ...
def __getitem__(self, index: int) -> T: ...
def __iter__(self): ...
class ReadOnlyList:
"""Does not claim to be a list. No inherited contract to violate."""
def __init__(self, data: list) -> None:
self._data = list(data) # defensive copy
def __len__(self) -> int:
return len(self._data)
def __getitem__(self, index):
return self._data[index]
def __iter__(self):
return iter(self._data)
def __repr__(self) -> str:
return f"ReadOnlyList({self._data!r})"
# Consumer only promises it needs something readable
def display(items: ReadableSequence) -> None:
for item in items:
print(item)
data = ReadOnlyList([1, 2, 3])
display(data) # ✓ works perfectly
LSP Checklist
| Rule | What to check |
|---|---|
| Preconditions | Subtype must not strengthen preconditions (accept at least what parent accepts) |
| Postconditions | Subtype must not weaken postconditions (return at least what parent promises) |
| Invariants | Subtype must maintain all invariants parent guarantees |
| Exception types | Subtype may only raise exceptions parent declares or subtypes thereof |
| Return types | Subtype may return a more specific type (covariance) |
| Parameter types | Subtype may accept a more general type (contravariance) |
Part 4 - Interface Segregation Principle (ISP)
Clients should not be forced to depend on interfaces they do not use.
Fat interfaces create coupling to methods the client doesn't call, and force implementors to stub out irrelevant methods.
The Violation - Fat StorageBackend
# isp/bad.py
from abc import ABC, abstractmethod
from typing import BinaryIO, Iterator
class StorageBackend(ABC):
"""Ten methods. Every implementor must implement all ten."""
@abstractmethod
def read(self, key: str) -> bytes: ...
@abstractmethod
def write(self, key: str, data: bytes) -> None: ...
@abstractmethod
def delete(self, key: str) -> None: ...
@abstractmethod
def exists(self, key: str) -> bool: ...
@abstractmethod
def list_keys(self, prefix: str = "") -> Iterator[str]: ...
@abstractmethod
def read_stream(self, key: str) -> BinaryIO: ...
@abstractmethod
def write_stream(self, key: str, stream: BinaryIO) -> None: ...
@abstractmethod
def copy(self, src: str, dst: str) -> None: ...
@abstractmethod
def move(self, src: str, dst: str) -> None: ...
@abstractmethod
def get_metadata(self, key: str) -> dict: ...
class AuditLogger(StorageBackend):
"""Only needs to READ, but must implement 9 other methods."""
def read(self, key: str) -> bytes:
# actual implementation
return open(key, "rb").read()
# Forced to implement every other method:
def write(self, key, data): raise NotImplementedError
def delete(self, key): raise NotImplementedError
def exists(self, key): raise NotImplementedError
def list_keys(self, prefix=""): raise NotImplementedError
def read_stream(self, key): raise NotImplementedError
def write_stream(self, key, stream): raise NotImplementedError
def copy(self, src, dst): raise NotImplementedError
def move(self, src, dst): raise NotImplementedError
def get_metadata(self, key): raise NotImplementedError
AuditLogger is coupled to eight methods it will never call or implement. Adding a new method to StorageBackend breaks AuditLogger.
Refactored - Composable Protocols
# isp/good.py
from __future__ import annotations
from typing import BinaryIO, Iterator, Protocol, runtime_checkable
# ── Small, focused protocols ───────────────────────────────────────────────────
@runtime_checkable
class Readable(Protocol):
def read(self, key: str) -> bytes: ...
def exists(self, key: str) -> bool: ...
@runtime_checkable
class Writable(Protocol):
def write(self, key: str, data: bytes) -> None: ...
@runtime_checkable
class Deletable(Protocol):
def delete(self, key: str) -> None: ...
class Listable(Protocol):
def list_keys(self, prefix: str = "") -> Iterator[str]: ...
class Streamable(Protocol):
def read_stream(self, key: str) -> BinaryIO: ...
def write_stream(self, key: str, stream: BinaryIO) -> None: ...
class Copyable(Protocol):
def copy(self, src: str, dst: str) -> None: ...
def move(self, src: str, dst: str) -> None: ...
# ── Compose only what you need ────────────────────────────────────────────────
class ReadWriteStorage(Readable, Writable, Protocol):
"""Compose for services that need read + write."""
class FullStorage(Readable, Writable, Deletable, Listable, Protocol):
"""Compose for services that need the full suite."""
# ── Implementors implement only what they support ─────────────────────────────
class LocalFileStorage:
"""Implements everything - satisfies all protocols structurally."""
def read(self, key: str) -> bytes:
with open(key, "rb") as f:
return f.read()
def write(self, key: str, data: bytes) -> None:
with open(key, "wb") as f:
f.write(data)
def delete(self, key: str) -> None:
import os
os.remove(key)
def exists(self, key: str) -> bool:
import os
return os.path.exists(key)
def list_keys(self, prefix: str = "") -> Iterator[str]:
import os
for name in os.listdir("."):
if name.startswith(prefix):
yield name
def read_stream(self, key: str) -> BinaryIO:
return open(key, "rb")
def write_stream(self, key: str, stream: BinaryIO) -> None:
with open(key, "wb") as f:
f.write(stream.read())
def copy(self, src: str, dst: str) -> None:
import shutil
shutil.copy2(src, dst)
def move(self, src: str, dst: str) -> None:
import shutil
shutil.move(src, dst)
class AuditLogger:
"""Only implements Readable. Not forced to stub anything else."""
def __init__(self, log_dir: str) -> None:
self._dir = log_dir
def read(self, key: str) -> bytes:
import os
path = os.path.join(self._dir, key)
with open(path, "rb") as f:
return f.read()
def exists(self, key: str) -> bool:
import os
return os.path.exists(os.path.join(self._dir, key))
# ── Consumers depend only on the protocols they actually use ──────────────────
class ReportGenerator:
"""Only needs to read - takes Readable, not FullStorage."""
def __init__(self, storage: Readable) -> None:
self._storage = storage
def generate(self, key: str) -> str:
data = self._storage.read(key)
return f"Report ({len(data)} bytes): {data[:80]!r}"
class BackupService:
"""Needs read + write + list."""
def __init__(self, src: Readable & Listable, dst: Writable) -> None: # type: ignore[operator]
self._src = src
self._dst = dst
def backup_all(self, prefix: str = "") -> int:
count = 0
for key in self._src.list_keys(prefix):
data = self._src.read(key)
self._dst.write(f"backup/{key}", data)
count += 1
return count
# Runtime check - works because @runtime_checkable
logger = AuditLogger("/var/log/audit")
assert isinstance(logger, Readable) # ✓
report = ReportGenerator(logger) # ✓ AuditLogger satisfies Readable
Protocol Composition Table
| Consumer | Protocols needed | Why not everything? |
|---|---|---|
ReportGenerator | Readable | Only reads, never writes |
AuditLogger | Readable | Append-only log, never deletes |
BackupService | Readable, Listable, Writable | Reads source, writes destination |
GarbageCollector | Listable, Deletable | Lists and removes stale files |
CDN Uploader | Readable, Streamable | Streams large files |
Part 5 - Dependency Inversion Principle (DIP)
High-level modules should not depend on low-level modules. Both should depend on abstractions.
The second half is equally important: Abstractions should not depend on details. Details should depend on abstractions.
The Violation - Hardwired Dependencies
# dip/bad.py
import smtplib
import sqlite3
from email.mime.text import MIMEText
class NotificationService:
"""High-level policy. Should not know about SMTP or SQLite."""
def __init__(self) -> None:
# Hardwired low-level details - impossible to test or swap
self._db = sqlite3.connect("notifications.db")
self._smtp_host = "smtp.gmail.com"
self._smtp_port = 587
def notify_user(self, user_id: int, message: str) -> None:
# Load email from SQLite - directly
row = self._db.execute(
"SELECT email FROM users WHERE id = ?", (user_id,)
).fetchone()
if not row:
raise ValueError(f"User {user_id} not found")
email = row[0]
# Send via SMTP - directly
msg = MIMEText(message)
msg["Subject"] = "Notification"
msg["To"] = email
with smtplib.SMTP(self._smtp_host, self._smtp_port) as server:
server.starttls()
server.login("user", "password")
server.send_message(msg)
Testing notify_user requires a real database and a live SMTP server. Swapping to SendGrid means editing NotificationService. The high-level module owns the low-level detail.
Refactored - Constructor Injection
# dip/good.py
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Protocol
# ── Abstractions (the stable layer both sides depend on) ──────────────────────
class UserLookup(Protocol):
def get_email(self, user_id: int) -> str: ...
class MessageSender(Protocol):
def send(self, to_email: str, subject: str, body: str) -> None: ...
# ── High-level policy (depends only on abstractions) ─────────────────────────
class NotificationService:
"""Does not know whether storage is SQLite, Postgres, or in-memory."""
def __init__(self, users: UserLookup, sender: MessageSender) -> None:
self._users = users
self._sender = sender
def notify_user(self, user_id: int, message: str) -> None:
email = self._users.get_email(user_id)
self._sender.send(email, "Notification", message)
# ── Low-level details (depend on abstractions, not the reverse) ───────────────
class SQLiteUserLookup:
def __init__(self, db_path: str) -> None:
import sqlite3
self._conn = sqlite3.connect(db_path)
def get_email(self, user_id: int) -> str:
row = self._conn.execute(
"SELECT email FROM users WHERE id = ?", (user_id,)
).fetchone()
if not row:
raise ValueError(f"User {user_id} not found")
return row[0]
class SMTPMessageSender:
def __init__(self, host: str, port: int, login: str, password: str) -> None:
self._host = host
self._port = port
self._login = login
self._password = password
def send(self, to_email: str, subject: str, body: str) -> None:
import smtplib
from email.mime.text import MIMEText
msg = MIMEText(body)
msg["Subject"] = subject
msg["To"] = to_email
with smtplib.SMTP(self._host, self._port) as server:
server.starttls()
server.login(self._login, self._password)
server.send_message(msg)
# ── In-memory fakes for testing (zero I/O) ───────────────────────────────────
class InMemoryUserLookup:
def __init__(self, users: dict[int, str]) -> None:
self._users = users
def get_email(self, user_id: int) -> str:
if user_id not in self._users:
raise ValueError(f"User {user_id} not found")
return self._users[user_id]
class RecordingMessageSender:
def __init__(self) -> None:
self.sent: list[dict] = []
def send(self, to_email: str, subject: str, body: str) -> None:
self.sent.append({"to": to_email, "subject": subject, "body": body})
# ── Manual wiring (composition root) ─────────────────────────────────────────
def build_production_service() -> NotificationService:
import os
return NotificationService(
users=SQLiteUserLookup(os.environ["DB_PATH"]),
sender=SMTPMessageSender(
host=os.environ["SMTP_HOST"],
port=int(os.environ.get("SMTP_PORT", "587")),
login=os.environ["SMTP_USER"],
password=os.environ["SMTP_PASSWORD"],
),
)
def build_test_service() -> tuple[NotificationService, RecordingMessageSender]:
sender = RecordingMessageSender()
svc = NotificationService(
sender=sender,
)
return svc, sender
# ── Test (pure Python, no I/O) ────────────────────────────────────────────────
def test_notify_user() -> None:
svc, sender = build_test_service()
svc.notify_user(1, "Your order shipped!")
assert len(sender.sent) == 1
assert "shipped" in sender.sent[0]["body"]
print("test_notify_user PASSED")
test_notify_user()
Using dependency-injector for Larger Applications
Manual wiring works well up to ~20 dependencies. Beyond that, a framework pays off.
# dip/container.py
from dependency_injector import containers, providers
from dip.good import (
NotificationService,
SQLiteUserLookup,
SMTPMessageSender,
)
class Container(containers.DeclarativeContainer):
config = providers.Configuration()
user_lookup = providers.Singleton(
SQLiteUserLookup,
db_path=config.db.path,
)
message_sender = providers.Singleton(
SMTPMessageSender,
host=config.smtp.host,
port=config.smtp.port,
login=config.smtp.login,
password=config.smtp.password,
)
notification_service = providers.Factory(
NotificationService,
users=user_lookup,
sender=message_sender,
)
# Bootstrap
container = Container()
container.config.from_yaml("config.yaml")
svc = container.notification_service()
svc.notify_user(42, "Hello from DI container!")
Part 6 - SOLID in the Real World: FastAPI Service Refactoring
Here is a complete before-and-after for a real FastAPI course enrollment service.
Before: SOLID-Violating FastAPI Handler
# before/main.py - everything in one place
from fastapi import FastAPI, HTTPException
import sqlite3
import smtplib
import hashlib
import re
import jwt
import time
app = FastAPI()
DB = "courses.db"
@app.post("/enroll")
def enroll(user_id: int, course_id: int, token: str):
# Inline auth
try:
payload = jwt.decode(token, "secret", algorithms=["HS256"])
except jwt.InvalidTokenError:
raise HTTPException(401, "Invalid token")
uid = payload.get("sub")
if str(uid) != str(user_id):
raise HTTPException(403, "Forbidden")
# Inline DB logic
conn = sqlite3.connect(DB)
row = conn.execute("SELECT * FROM enrollments WHERE user_id=? AND course_id=?",
(user_id, course_id)).fetchone()
if row:
raise HTTPException(409, "Already enrolled")
conn.execute("INSERT INTO enrollments VALUES (?,?)", (user_id, course_id))
conn.commit()
# Inline email
try:
server = smtplib.SMTP("smtp.example.com")
f"You enrolled in course {course_id}")
server.quit()
except Exception:
pass # swallow email errors
return {"status": "enrolled"}
Issues: auth, DB, email, business rules - all tangled. Untestable without real SMTP and SQLite. No separation of concerns.
After: SOLID-Compliant FastAPI Service
# after/domain.py
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Protocol
import re
@dataclass(frozen=True)
class UserId:
value: int
@dataclass(frozen=True)
class CourseId:
value: int
@dataclass(frozen=True)
class EmailAddress:
value: str
def __post_init__(self) -> None:
if not re.match(r"[^@]+@[^@]+\.[^@]+", self.value):
raise ValueError(f"Invalid email: {self.value!r}")
class EnrollmentRepository(Protocol):
def is_enrolled(self, user_id: UserId, course_id: CourseId) -> bool: ...
def enroll(self, user_id: UserId, course_id: CourseId) -> None: ...
class NotificationPort(Protocol):
def send_enrollment_confirmation(
self, email: EmailAddress, course_id: CourseId
) -> None: ...
class TokenVerifier(Protocol):
def verify(self, token: str) -> dict: ...
# after/application.py
from __future__ import annotations
from dataclasses import dataclass
from after.domain import (
UserId, CourseId, EmailAddress,
EnrollmentRepository, NotificationPort, TokenVerifier,
)
class AlreadyEnrolledError(Exception):
pass
class EnrollmentService:
"""Single responsibility: orchestrate the enrollment use case."""
def __init__(
self,
repo: EnrollmentRepository,
notifier: NotificationPort,
token_verifier: TokenVerifier,
) -> None:
self._repo = repo
self._notifier = notifier
self._verifier = token_verifier
def enroll(self, user_id: int, course_id: int, token: str) -> None:
payload = self._verifier.verify(token) # raises on invalid
uid = UserId(user_id)
cid = CourseId(course_id)
email = EmailAddress(payload["email"])
if self._repo.is_enrolled(uid, cid):
raise AlreadyEnrolledError(f"User {user_id} already enrolled in {course_id}")
self._repo.enroll(uid, cid)
self._notifier.send_enrollment_confirmation(email, cid)
# after/adapters/sqlite_repo.py
import sqlite3
from after.domain import UserId, CourseId, EnrollmentRepository
class SQLiteEnrollmentRepository:
def __init__(self, db_path: str) -> None:
self._db = db_path
with sqlite3.connect(db_path) as conn:
conn.execute(
"CREATE TABLE IF NOT EXISTS enrollments "
"(user_id INTEGER, course_id INTEGER, PRIMARY KEY (user_id, course_id))"
)
def is_enrolled(self, user_id: UserId, course_id: CourseId) -> bool:
with sqlite3.connect(self._db) as conn:
row = conn.execute(
"SELECT 1 FROM enrollments WHERE user_id=? AND course_id=?",
(user_id.value, course_id.value),
).fetchone()
return row is not None
def enroll(self, user_id: UserId, course_id: CourseId) -> None:
with sqlite3.connect(self._db) as conn:
conn.execute(
"INSERT INTO enrollments VALUES (?, ?)",
(user_id.value, course_id.value),
)
# after/adapters/smtp_notifier.py
import smtplib
from after.domain import EmailAddress, CourseId, NotificationPort
class SMTPNotificationAdapter:
def __init__(self, smtp_host: str, sender: str) -> None:
self._host = smtp_host
self._sender = sender
def send_enrollment_confirmation(
self, email: EmailAddress, course_id: CourseId
) -> None:
with smtplib.SMTP(self._host) as server:
server.sendmail(
self._sender,
email.value,
f"Subject: Enrolled\n\nYou enrolled in course {course_id.value}",
)
# after/adapters/jwt_verifier.py
import jwt
from after.domain import TokenVerifier
class JWTTokenVerifier:
def __init__(self, secret: str, algorithm: str = "HS256") -> None:
self._secret = secret
self._algorithm = algorithm
def verify(self, token: str) -> dict:
return jwt.decode(token, self._secret, algorithms=[self._algorithm])
# after/api.py
from fastapi import FastAPI, HTTPException, Header
from after.application import EnrollmentService, AlreadyEnrolledError
from after.adapters.sqlite_repo import SQLiteEnrollmentRepository
from after.adapters.smtp_notifier import SMTPNotificationAdapter
from after.adapters.jwt_verifier import JWTTokenVerifier
import jwt as _jwt
app = FastAPI()
# Composition root - wire everything together once
_repo = SQLiteEnrollmentRepository("courses.db")
_verifier = JWTTokenVerifier("your-secret")
_service = EnrollmentService(_repo, _notifier, _verifier)
@app.post("/enroll")
def enroll(
user_id: int,
course_id: int,
authorization: str = Header(...),
):
token = authorization.removeprefix("Bearer ")
try:
_service.enroll(user_id, course_id, token)
except _jwt.InvalidTokenError:
raise HTTPException(401, "Invalid token")
except PermissionError:
raise HTTPException(403, "Forbidden")
except AlreadyEnrolledError as e:
raise HTTPException(409, str(e))
return {"status": "enrolled"}
# after/tests/test_enrollment.py
import pytest
import jwt
import time
from after.application import EnrollmentService, AlreadyEnrolledError
from after.domain import UserId, CourseId, EmailAddress
# ── In-memory fakes ───────────────────────────────────────────────────────────
class FakeEnrollmentRepo:
def __init__(self):
self.enrollments: set[tuple[int, int]] = set()
def is_enrolled(self, user_id: UserId, course_id: CourseId) -> bool:
return (user_id.value, course_id.value) in self.enrollments
def enroll(self, user_id: UserId, course_id: CourseId) -> None:
self.enrollments.add((user_id.value, course_id.value))
class FakeNotifier:
def __init__(self):
self.sent: list[tuple[str, int]] = []
def send_enrollment_confirmation(self, email: EmailAddress, course_id: CourseId):
self.sent.append((email.value, course_id.value))
class FakeTokenVerifier:
def __init__(self, payload: dict):
self._payload = payload
def verify(self, token: str) -> dict:
return self._payload
# ── Tests (no I/O whatsoever) ─────────────────────────────────────────────────
def make_service():
repo = FakeEnrollmentRepo()
notifier = FakeNotifier()
svc = EnrollmentService(repo, notifier, verifier)
return svc, repo, notifier
def test_successful_enrollment():
svc, repo, notifier = make_service()
svc.enroll(1, 42, "fake-token")
assert repo.is_enrolled(UserId(1), CourseId(42))
assert len(notifier.sent) == 1
def test_duplicate_enrollment_raises():
svc, repo, _ = make_service()
svc.enroll(1, 42, "fake-token")
with pytest.raises(AlreadyEnrolledError):
svc.enroll(1, 42, "fake-token")
def test_notification_sent_on_enrollment():
svc, _, notifier = make_service()
svc.enroll(1, 99, "fake-token")
assert notifier.sent[0][1] == 99
The tests run in milliseconds with no real database, no real SMTP, and no real JWT library - because every dependency is inverted through a protocol.
Interview Patterns
SOLID principles appear constantly in senior engineer and staff engineer interviews. Here are the patterns that come up most.
Pattern 1 - "What does SRP actually mean?"
Interviewers test whether you regurgitate "one method" or give the real answer.
Strong answer: "SRP is about cohesion and stakeholder ownership. A class has a single responsibility when there is one team or person who could reasonably request a change to it. The signal is: if two different stakeholders can independently ask you to modify the same class, it has too many responsibilities."
Pattern 2 - "How do you extend behaviour without modifying existing code?"
Strong answer: Demonstrate the strategy registry or singledispatch pattern. Show that OCP doesn't mean classes are frozen - it means extension points are designed upfront so that adding new behaviour doesn't require touching proven code.
Pattern 3 - "Explain an LSP violation you've encountered in production"
Common real examples:
MockDBthat inherits real DB class but raisesNotImplementedErroron write methodsRestrictedListthat inheritslistbut prevents mutationAdminUserthat inheritsUserbut overridescan_accessto returnTruefor everything, breaking audit logic
Strong answer: Frame it as a contract violation, not just a "behaviour change."
Pattern 4 - "How do you test a class that sends emails and writes to a database?"
Strong answer: "I wouldn't design it that way. By applying DIP, the service depends on protocols rather than concrete SMTP and database classes. In tests I inject in-memory fakes. The test runs in microseconds with no network or disk I/O."
Pattern 5 - "Design a plugin system"
This combines OCP and DIP. A good answer:
- Define a
Pluginprotocol (the abstraction) - Use a registry to discover plugins at runtime (
@register_plugindecorator or entry points) - The host application depends on the protocol, never on concrete plugin classes
- New plugins are added as new modules - zero changes to host code
Quick Reference Table
| Principle | One-line | Signal of violation |
|---|---|---|
| SRP | One reason to change | Multiple stakeholders own one class |
| OCP | Extend by adding, not editing | New feature requires if/elif in existing code |
| LSP | Subtypes are drop-in replacements | isinstance checks in calling code |
| ISP | Interfaces match their consumers | Implementors stub methods with raise NotImplementedError |
| DIP | Depend on abstractions, not concretions | __init__ calls sqlite3.connect or smtplib.SMTP directly |
The SOLID Smell Detector
# Run this mental checklist against any class you're reviewing:
solid_smells = {
"SRP": [
"More than one import domain (e.g., db + smtp + crypto)",
"Class name contains 'Manager', 'Handler', 'Helper', 'Utils'",
"Test requires mocking more than 2 external systems",
],
"OCP": [
"Adding a new variant requires editing an existing function",
"Long if/elif/elif chain dispatching on type or string",
"Switch statement on an enum that grows over time",
],
"LSP": [
"Subclass raises NotImplementedError on parent methods",
"Calling code has isinstance() checks after receiving a base type",
"Subclass narrows parameter types or widens exception types",
],
"ISP": [
"ABC with 8+ abstract methods",
"Implementors stub most methods with 'pass' or 'raise NotImplementedError'",
"Tests mock a large interface to use 2 methods",
],
"DIP": [
"Class instantiates its dependencies in __init__",
"Hard-coded connection strings or API keys inside business logic",
"Impossible to test without real network or disk",
],
}
